package com.hmdp.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.json.JSONUtil;
import com.hmdp.dto.Result;
import com.hmdp.entity.VoucherOrder;
import com.hmdp.mapper.VoucherOrderMapper;
import com.hmdp.service.ISeckillVoucherService;
import com.hmdp.service.IVoucherOrderService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hmdp.service.IVoucherService;
import com.hmdp.utils.RedisIdWorker;
import com.hmdp.utils.UserHolder;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.aop.framework.AopContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * <p>
 *  服务实现类
 * </p>
 *
 * @author 虎哥
 * @since 2021-12-22
 */
@Service
@Slf4j
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {
    @Autowired
    private IVoucherService voucherService;
    @Autowired
    private ISeckillVoucherService seckillVoucherService;
    @Autowired
    private RedisIdWorker redisIdWorker;
    @Autowired
    private RedissonClient redissonClient;
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    //基于jvm内存的阻塞队列
    private static final BlockingQueue orderTasks = new ArrayBlockingQueue(1024*1024);
    private final static DefaultRedisScript<Long> SECKILL;
    static {
        SECKILL = new DefaultRedisScript<>();
        SECKILL.setLocation(new ClassPathResource("/lua/seckill.lua"));
        SECKILL.setResultType(Long.class);
    }
    //创建线程池，执行下单任务
    private static final ExecutorService ORDER_TASK_EXECUTOR = Executors.newSingleThreadExecutor();
    private static  IVoucherOrderService proxy;
    @PostConstruct
    public void init() {
        //子线程无法获取父线程的代理对象，因此需要定义属性获取
        ORDER_TASK_EXECUTOR.submit(new orderTaskHandler());
    }
    class orderTaskHandler implements Runnable {
        String queueName="stream.orders";
        @Override
        public void run() {
            while (true){
                try {
                    //XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS stream.orders >
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(Consumer.from("g1","c1"),
                            StreamReadOptions.empty().count(1).block(Duration.ofMillis(2)),
                            StreamOffset.create(queueName, ReadOffset.lastConsumed()));
                    //每次处理一个
                    if (list==null||list.isEmpty()){
                        //获取失败，当前没有消息要处理
                        continue;
                    }
                    MapRecord<String, Object, Object> entries = list.get(0);
                    Map<Object, Object> orderMap = entries.getValue();
                    //这里由于VoucherOrder添加了builder注解，覆盖了无参构造器，因此需要手动提供一个
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(orderMap, new VoucherOrder(), true);
                    handleVoucherOrder(voucherOrder);
                    //ack
                    stringRedisTemplate.opsForStream().acknowledge(queueName,"g1",entries.getId());
                } catch (Exception e) {
                    log.error("获取阻塞队列出现问题",e);
                    //从pending list获取数据
                    handlePendingList();
                }
            }
        }

        private void handlePendingList() {
            while (true){
                try {
                    //XREADGROUP GROUP g1 c1 COUNT 1  STREAMS stream.orders 0
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(Consumer.from("g1","c1"),
                            StreamReadOptions.empty().count(1),
                            StreamOffset.create(queueName, ReadOffset.from("0")));
                    //每次处理一个
                    if (list==null||list.isEmpty()){
                        //获取失败，当前没有消息要处理
                        break;
                    }
                    MapRecord<String, Object, Object> entries = list.get(0);
                    Map<Object, Object> orderMap = entries.getValue();
                    //这里由于VoucherOrder添加了builder注解，覆盖了无参构造器，因此需要手动提供一个
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(orderMap, new VoucherOrder(), true);
                    handleVoucherOrder(voucherOrder);
                    //ack  xack steam.orders g1 id
                    stringRedisTemplate.opsForStream().acknowledge(queueName,"g1",entries.getId());
                } catch (Exception e) {
                    log.error("获取阻塞队列出现问题",e);
                    //出故障后过一会再处理
                    try {
                        Thread.sleep(2000);
                    } catch (InterruptedException ex) {
                        throw new RuntimeException(ex);
                    }
                    continue;
                }
            }
        }
    }
    private void handleVoucherOrder(VoucherOrder voucherOrder) {
        Long userId = voucherOrder.getUserId();
        RLock redisLock = redissonClient.getLock("lock:order:"+userId);
        //无参表示失败不等待
        //这里其实锁是为了保底，redis已经实现了一人一单的判断了，确实不需要再上锁，只是为了防止redis崩了
        boolean isLock = redisLock.tryLock();
        if (!isLock){
            //获取锁失败
            log.error("请勿重复下单喔");
//            return Result.fail("请勿重复下单喔");
        }
        try {
            proxy.creatSeckillVoucher(voucherOrder);
        } finally {
            //释放锁
            redisLock.unlock();
        }
    }
    @Override
    /**
     * 基于redis的stream结构实现消息队列
     */
    public Result seckillVoucher(Long voucherId) {
        //执行lua
        Long userId = UserHolder.getUser().getId();
        //orderId
        Long orderId = redisIdWorker.nextID("order");

        Long flag = stringRedisTemplate.execute(SECKILL, Collections.emptyList(), voucherId.toString(), userId.toString(),orderId.toString());
        //下单失败
        if (flag!=0){
            if (flag.intValue()==1){
                return Result.fail("库存不足，下次请早");
            }else if (flag.intValue()==2){
                return Result.fail("请勿重复下单！");
            }
        }
        //下单成功
        //无需再往消息队列中存入voucherOrder对象
        //获取代理对象
        proxy = (IVoucherOrderService) AopContext.currentProxy();
        return Result.ok(orderId);
    }
    //创建线程对象
//    class orderTaskHandler implements Runnable {
//
//        @Override
//        public void run() {
//            while (true){
//                //获取阻塞队列中的值
//                try {
//                    VoucherOrder order = (VoucherOrder) orderTasks.take();
//                    handleVoucherOrder(order);
//                } catch (Exception e) {
//                    log.error("获取阻塞队列出现问题",e);
//                }
//            }
//        }
//    }
//    private void handleVoucherOrder(VoucherOrder voucherOrder) {
//        Long userId = voucherOrder.getUserId();
//        RLock redisLock = redissonClient.getLock("lock:order:"+userId);
//        //无参表示失败不等待
//        //这里其实锁是为了保底，redis已经实现了一人一单的判断了，确实不需要再上锁，只是为了防止redis崩了
//        boolean isLock = redisLock.tryLock();
//        if (!isLock){
//            //获取锁失败
//            log.error("请勿重复下单喔");
////            return Result.fail("请勿重复下单喔");
//        }
//        try {
//            proxy.creatSeckillVoucher(voucherOrder);
//        } finally {
//            //释放锁
//            redisLock.unlock();
//        }
//    }
//    @Override
//    /**
//     * 改为由redis实现秒杀库存以及一人一单的判断
//     */
//    public Result seckillVoucher(Long voucherId) {
//        //执行lua
//        Long userId = UserHolder.getUser().getId();
//        //orderId
//        Long orderId = redisIdWorker.nextID("order");
//
//        Long flag = stringRedisTemplate.execute(SECKILL, Collections.emptyList(), voucherId.toString(), userId.toString(),orderId.toString());
//        //下单失败
//        if (flag!=0){
//            if (flag.intValue()==1){
//                return Result.fail("库存不足，下次请早");
//            }else if (flag.intValue()==2){
//                return Result.fail("请勿重复下单！");
//            }
//        }
//        //下单成功
//        //创建订单，结果存入阻塞队列
//        VoucherOrder voucherOrder = VoucherOrder.builder()
//                .id(orderId)
//                .userId(userId)
//                .voucherId(voucherId)
//                .build();
//        orderTasks.add(voucherOrder);
//        //获取代理对象
//        proxy = (IVoucherOrderService) AopContext.currentProxy();
//        return Result.ok(orderId);
//    }
    @Transactional
    public void creatSeckillVoucher(VoucherOrder voucherOrder) {
        //判断是否已经下过单
        //因为是独立子线程，因此无法通过userHolder获取userId
        Long userId = voucherOrder.getUserId();
        Long voucherId = voucherOrder.getVoucherId();
        Integer count = query().eq("user_id", userId).eq("voucher_id",voucherId ).count();
        if (count>0){
//            return Result.fail("每人只能抢一单喔");
            log.error("每人只能抢一单喔");
        }
        //库存足，插数据到voucherOrder表
        boolean flag = seckillVoucherService.update()
                .setSql("stock=stock-1").eq("voucher_id", voucherId).gt("stock", 0)
                .update();
        if (!flag){
//            return Result.fail("您的手速不够快");
            log.error("您的手速不够快");
        }
        //返回订单id
        save(voucherOrder);
        return ;
    }
//    @Override
//    public Result seckillVoucher(Long voucherId) {
//        //基于voucherId查优惠券信息
//        SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
//        //判断秒杀是否未开始或结束
//        //是，返回
//        LocalDateTime now = LocalDateTime.now();
//        if (voucher.getBeginTime().isAfter(now)){
//            return Result.fail("尚未到优惠券抢购时间");
//        }
//        if (voucher.getEndTime().isBefore(now)){
//            return Result.fail("已过抢购时间，下次请早");
//        }
//
//        //否，判断库存
//        Integer stock = voucher.getStock();
//        if (stock<1){
//            //库存不足
//            return Result.fail("库存不足，下次请早");
//        }
//        Long id = UserHolder.getUser().getId();
//        //这里希望，为相同的id上锁，确保一人一单，但是synchronized检测的对象，即地址，而tostring都是new string
////        synchronized (id.toString().intern()){
////            //同类方法中直接调用，获取的是目标对象，而不是代理对象，Transactional注解就失效
////            //因此要获得当前接口的代理对象
////            IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
////            return proxy.creatSeckillVoucher(voucherId);
////        }
//        String key = "order:"+id;
////        SimpleRedisLock redisLock = new SimpleRedisLock(stringRedisTemplate,key);
//        //使用redisson提供的锁替代自己实现的，包括了trylock和unlock
//        RLock redisLock = redissonClient.getLock("anyLock");
//        //无参表示失败不等待
//        boolean isLock = redisLock.tryLock();
//        if (!isLock){
//            //获取锁失败
//            return Result.fail("请勿重复下单喔");
//        }
//        try {
//            IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
//            return proxy.creatSeckillVoucher(voucherId);
//        } finally {
//            //释放锁
//            redisLock.unlock();
//        }
//    }
//    @Transactional
//    public Result creatSeckillVoucher(Long voucherId) {
//        //判断是否已经下过单
//        Long id = UserHolder.getUser().getId();
//        Integer count = query().eq("user_id", id).eq("voucher_id", voucherId).count();
//        if (count>0){
//            return Result.fail("每人只能抢一单喔");
//        }
//        //库存足，插数据到voucherOrder表
//        boolean flag = seckillVoucherService.update()
//                .setSql("stock=stock-1").eq("voucher_id", voucherId).gt("stock", 0)
//                .update();
//        if (!flag){
//            return Result.fail("您的手速不够快");
//        }
//        long orderid = redisIdWorker.nextID("order");
//        VoucherOrder voucherOrder = VoucherOrder.builder()
//                .id(orderid)
//                .userId(id)
//                .voucherId(voucherId)
//                .build();
//        save(voucherOrder);
//        //返回订单id
//        return Result.ok(id);
//    }
}
